// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/base/feed/ws/kucoin_subscriber.h"
#include "coin2/exchange/base/feed/ws/kucoin_connection.h"

#include <memory>

#include <boost/asio/spawn.hpp>

#include "coin2/base/log.h"

namespace coin2::exchange::base::feed {

using coin2::exchange::feed::WebsocketConfig;

void KucoinSubscriber::ConnectForSymbolGroupAndWorker(
    const WsConnectionConfig* connection_config) {
  std::weak_ptr<WsSubscriber> self(shared_from_this());
  std::shared_ptr<LatencyExtractor> extractor(nullptr);
  if (config_.reconnect_slowest()) {
    extractor = latency_manager_.AddLatencyExtractor(connection_config);
  }
  std::shared_ptr<WsConnection> conn = std::make_shared<KucoinConnection>(
      connection_config,
      [self, connection_config] {
        if (auto s = self.lock()) {
          boost::asio::post(
              s->strand_, [self, connection_config] {
                if (auto s = self.lock()) {
                  s->latency_manager_.RemoveLatencyExtractor(connection_config);
                  s->AddForSymbolGroupAndWorker(connection_config);
                } else {
                  LOG(ERROR)
                      << "boost::asio handler object (KucoinSubscriber) does "
                         "not exist";
                }
              });
        } else {
          LOG(ERROR)
              << "boost::asio handler object (KucoinSubscriber) does not exist";
        }
      },
      flow_sub_, request_generator_.get(), extractor, enable_telemetry_);
  conn->RunAsync();
  {
    std::scoped_lock _(connections_mutex_);
    ConnectionKey k(connection_config->connection_id,
                    connection_config->worker_id);
    connections_.insert({k, conn});
  }
}
}  // namespace coin2::exchange::base::feed
